// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
//
// Author: wsfuyibing <682805@qq.com>
// Date: 2024-11-30

package managers

import (
	"context"
	"fmt"
	"gitee.com/go-libs/log"
	"gitee.com/go-libs/process"
	"gitee.com/gomq/sr"
	"gomq/app/errs"
	"gomq/app/models"
	"gomq/app/services"
	"sync"
	"time"
)

const (
	memoryReloadFrequency = time.Minute
)

type (
	// MemoryManager is a component used to store relations data in
	// memory.
	MemoryManager interface {
		Attach(name string, notifier MemoryNotifier)
		BuildSubscription(task *models.Task) sr.Subscription
		Detach(name string)
		GetCluster() (cluster *models.Cluster, err error)
		GetClusterModel(id int) (cluster *models.Cluster, has bool)
		GetClusterModels() (clusters map[int]*models.Cluster)
		GetSubscription(id int) (subscription sr.Subscription, has bool)
		GetTaskModel(id int) (task *models.Task, has bool)
		GetTaskModels() (tasks map[int]*models.Task)
		HasRegistry(topic sr.TopicName, tag sr.TopicTag) (has bool)
		Load(ctx context.Context) (err error)
		Process() (process process.Process)
	}

	// MemoryNotifier is a handler that called when data in memory
	// update succeed.
	MemoryNotifier func(ctx context.Context) (err error)

	memory struct {
		manager   Manager
		mu        *sync.RWMutex
		name      string
		notifiers map[string]MemoryNotifier
		process   process.Process

		loadedCluster      map[int]*models.Cluster
		loadedRegistry     map[string]bool
		loadedSubscription map[int]sr.Subscription
		loadedTask         map[int]*models.Task
		loaderProcessing   bool
		loaderTime         time.Time
	}
)

func (o *memory) Attach(name string, notifier MemoryNotifier) {
	o.mu.Lock()
	defer o.mu.Unlock()
	if _, ok := o.notifiers[name]; ok {
		return
	}
	o.notifiers[name] = notifier
}

func (o *memory) BuildSubscription(task *models.Task) sr.Subscription {
	return o.buildSubscription(task)
}

func (o *memory) Detach(name string) {
	o.mu.Lock()
	defer o.mu.Unlock()
	if _, ok := o.notifiers[name]; ok {
		delete(o.notifiers, name)
	}
}

func (o *memory) GetCluster() (cluster *models.Cluster, err error) {
	o.mu.RLock()
	defer o.mu.RUnlock()

	// There is at least one record in the memory.
	if count := len(o.loadedCluster); count > 0 {
		// Return error
		// if too many clusters are enabled.
		if count > 1 {
			err = errs.ErrClusterTooMany
			return
		}
		for _, cluster = range o.loadedCluster {
			break
		}
		return
	}

	// Return error
	// if no enabled cluster.
	err = errs.ErrClusterNotFound
	return
}

func (o *memory) GetClusterModel(id int) (cluster *models.Cluster, exists bool) {
	o.mu.RLock()
	cluster, exists = o.loadedCluster[id]
	o.mu.RUnlock()
	return
}

func (o *memory) GetClusterModels() (clusters map[int]*models.Cluster) {
	o.mu.RLock()
	clusters = o.loadedCluster
	o.mu.RUnlock()
	return
}

func (o *memory) GetSubscription(id int) (subscription sr.Subscription, exists bool) {
	o.mu.RLock()
	subscription, exists = o.loadedSubscription[id]
	o.mu.RUnlock()
	return
}

func (o *memory) GetTaskModel(id int) (task *models.Task, exists bool) {
	o.mu.RLock()
	task, exists = o.loadedTask[id]
	o.mu.RUnlock()
	return
}

func (o *memory) GetTaskModels() (tasks map[int]*models.Task) {
	o.mu.RLock()
	tasks = o.loadedTask
	o.mu.RUnlock()
	return
}

func (o *memory) HasRegistry(topic sr.TopicName, tag sr.TopicTag) (has bool) {
	if _, ok := o.loadedRegistry[o.buildRegistryKey(topic, tag)]; ok {
		has = true
	}
	return
}

func (o *memory) Load(ctx context.Context) (err error) {
	return o.onLoad(ctx)
}

func (o *memory) Process() process.Process {
	return o.process
}

// + Process

func (o *memory) Run(ctx context.Context) (err error) {
	var ticker = time.NewTicker(memoryReloadFrequency)
	for {
		select {
		case <-ticker.C:
			go func() { _ = o.onLoad(ctx) }()
		case <-ctx.Done():
			ticker.Stop()
			return
		}
	}
}

// + Loads

func (o *memory) onLoad(ctx context.Context) (err error) {
	o.mu.Lock()

	// Return if loader processing in other goroutine.
	if o.loaderProcessing {
		o.mu.Unlock()
		return
	}

	// Update locked status.
	tc, tm := time.Now(), o.loaderTime
	o.loaderProcessing = true
	o.mu.Unlock()

	// Triggered when process done.
	defer func() {
		o.mu.Lock()
		o.loaderProcessing = false

		// Load failed.
		if err != nil {
			o.mu.Unlock()
			return
		}

		// Load succeed.
		o.loaderTime = tc
		notifiers := o.notifiers
		o.mu.Unlock()

		// Call notifiers in goroutines if success loaded.
		for _, x := range notifiers {
			go func(notifier MemoryNotifier) {
				_ = notifier(ctx)
			}(x)
		}
	}()

	// Sequential loaders.
	for _, loader := range []func(context.Context, time.Time) error{
		o.onLoadCluster,
		o.onLoadTask,
	} {
		if err = loader(ctx, tm); err != nil {
			break
		}
	}
	return
}

func (o *memory) onLoadCluster(ctx context.Context, tm time.Time) (err error) {
	var (
		count int
		list  []*models.Cluster
		reset = func(item *models.Cluster) {
			o.mu.Lock()
			defer o.mu.Unlock()
			if item.Status.Enabled() {
				if v, ok := o.loadedCluster[item.Id]; ok && v.UpdatedAt.Time().Unix() == item.UpdatedAt.Time().Unix() {
					return
				}
				o.loadedCluster[item.Id] = item
			} else {
				if _, ok := o.loadedCluster[item.Id]; ok {
					delete(o.loadedCluster, item.Id)
				}
			}
		}
		total = func() int {
			o.mu.RLock()
			defer o.mu.RUnlock()
			return len(o.loadedCluster)
		}
	)

	// Trigger when cluster load done.
	defer func() {
		if r := recover(); r != nil {
			log.Fatalf(`[%s] loader caused an panic: %v`, o.name, r)

			if err == nil {
				err = fmt.Errorf(`%v`, r)
			}
		}

		// Logger result.
		log.Infof(`[%s] cluster models loaded: updated-count="%d", loaded-count="%d"`, o.name, count, total())
	}()

	// Read items from database.
	if list, err = services.NewClusterService().LatestUpdates(ctx, tm); err != nil {
		return
	}

	// Do nothing if no item updated since last.
	if count = len(list); count == 0 {
		return
	}

	// Iterate items in database.
	for _, item := range list {
		reset(item)
	}
	return
}

func (o *memory) onLoadTask(ctx context.Context, tm time.Time) (err error) {
	var (
		count int
		list  []*models.Task
		reset = func(item *models.Task) {
			o.mu.Lock()
			defer o.mu.Unlock()

			key := o.buildRegistryKey(item.Topic, item.Tag)
			o.loadedRegistry[key] = true

			if item.Status.Enabled() {
				if v, ok := o.loadedTask[item.Id]; ok && v.UpdatedAt.Time().Unix() == item.UpdatedAt.Time().Unix() {
					return
				}
				o.loadedSubscription[item.Id] = o.buildSubscription(item)
				o.loadedTask[item.Id] = item
				return
			}

			if _, ok := o.loadedTask[item.Id]; ok {
				delete(o.loadedSubscription, item.Id)
				delete(o.loadedTask, item.Id)
			}
		}
		total = func() int {
			o.mu.RLock()
			n := len(o.loadedTask)
			o.mu.RUnlock()
			return n
		}
	)

	// Trigger when cluster load done.
	defer func() {
		if r := recover(); r != nil {
			log.Fatalf(`[%s] loader caused an panic: %v`, o.name, r)

			if err == nil {
				err = fmt.Errorf(`%v`, r)
			}
		}

		// Logger result.
		log.Infof(`[%s] task models loaded: updated-count="%d", loaded-count="%d"`, o.name, count, total())
	}()

	// Read items from database.
	if list, err = services.NewTaskService().LatestUpdates(ctx, tm); err != nil {
		return
	}

	// Do nothing if no item updated since last.
	if count = len(list); count == 0 {
		return
	}

	// Iterate items in database.
	for _, item := range list {
		reset(item)
	}
	return
}

// + Access

func (o *memory) buildRegistryKey(topic sr.TopicName, tag sr.TopicTag) string {
	return fmt.Sprintf(`%s:%s`,
		topic.ToLower(),
		tag.ToLower(),
	)
}

func (o *memory) buildSubscription(task *models.Task) (s sr.Subscription) {
	// Basic attributes setup.
	s = sr.NewSubscription().
		SetId(task.Id).SetTopic(task.Topic).SetTag(task.Tag).
		SetConcurrency(task.Concurrency).SetThreads(task.Threads).
		SetMaxRetry(task.MaxRetry).SetDelaySeconds(task.DelaySeconds).
		SetStore(task.Store.Enabled())

	// Filter for received message body.
	if !task.ConditionFormat.Undefined() && !task.ConditionExpression.Undefined() {
		s.SetFilter(
			sr.NewFilter(
				task.ConditionFormat,
				task.ConditionExpression,
			),
		)
	}

	// Endpoint of the subscription setup.
	if !task.Handler.Undefined() {
		ep := sr.NewEndpoint().
			SetAddress(task.Handler).
			SetFormat(task.HandlerFormat).
			SetMethod(task.HandlerMethod).
			SetTimeout(task.HandlerTimeout)

		// Filter for response body.
		if !task.HandlerResponseFormat.Undefined() && !task.HandlerResponseExpression.Undefined() {
			ep.SetFilter(sr.NewFilter(task.HandlerResponseFormat, task.HandlerResponseExpression))
		}

		// Bind endpoint.
		s.WithEndpoint(ep)
	}

	// Failed notification setup.
	if !task.Failed.Undefined() {
		ep := sr.NewEndpoint().
			SetAddress(task.Failed).
			SetFormat(task.FailedFormat).
			SetMethod(task.FailedMethod).
			SetTimeout(task.FailedTimeout)

		// Filter for response body.
		if !task.FailedResponseFormat.Undefined() && !task.FailedResponseExpression.Undefined() {
			ep.SetFilter(sr.NewFilter(task.FailedResponseFormat, task.FailedResponseExpression))
		}

		// Bind failed notification.
		s.WithFailedNotification(ep)
	}

	// Succeed notification setup.
	if !task.Succeed.Undefined() {
		ep := sr.NewEndpoint().
			SetAddress(task.Succeed).
			SetFormat(task.SucceedFormat).
			SetMethod(task.SucceedMethod).
			SetTimeout(task.SucceedTimeout)

		// Filter for response body.
		if !task.SucceedResponseFormat.Undefined() && !task.SucceedResponseExpression.Undefined() {
			ep.SetFilter(sr.NewFilter(task.SucceedResponseFormat, task.SucceedResponseExpression))
		}

		// Bind succeed notification.
		s.WithSucceedNotification(ep)
	}

	// Done.
	return s
}

// + Constructor

func (o *memory) init() *memory {
	o.name = "memory-manager"
	o.loadedCluster = make(map[int]*models.Cluster)
	o.loadedRegistry = make(map[string]bool)
	o.loadedSubscription = make(map[int]sr.Subscription)
	o.loadedTask = make(map[int]*models.Task)
	o.loaderTime = time.Unix(0, 0)
	o.mu = new(sync.RWMutex)
	o.notifiers = make(map[string]MemoryNotifier)
	o.process = process.New(o.name, o)
	return o
}
